package com.ych.service;

import com.alibaba.fastjson.JSON;
import com.ych.beans.Alarm;
import com.ych.config.Configure;
import org.apache.commons.lang.StringUtils;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
import org.springframework.context.ApplicationContext;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.core.ProducerFactory;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

import javax.websocket.*;
import javax.websocket.server.ServerEndpoint;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CopyOnWriteArraySet;

@ServerEndpoint(value = "/monitor")
@Component
public class MonitorServiceImpl {
    static List<String> alarmList = new ArrayList<>();

    // concurrent包的线程安全Set，用来存放每个客户端对应的Session对象。
    static CopyOnWriteArraySet<Session> SessionSet = new CopyOnWriteArraySet<Session>();

    @OnOpen//连接建立成功自动调用的方法,注意，这里的session不是HttpSession,是javax.websocket包中的session
    public void onOpen(Session session) {
        System.out.println("onOpen方法执行,sessionid="+session.getId());
        SessionSet.add(session);

        //向新加入连接发送所有历史报警数据
        SendMessage(session, alarmList.toString());
    }

    @KafkaListener(topics="alarm",groupId = "gr02")
    public void newAlarm(ConsumerRecord<String,String> r, Acknowledgment ack) throws IOException {
        Alarm alarm = JSON.parseObject(r.value(), Alarm.class);
        //如果存在该传感器的报警则删除并写入新的
        alarmList.removeIf(a -> JSON.parseObject(a, Alarm.class).getSensor().getId().equals(alarm.getSensor().getId()));
        alarmList.add(r.value());

        //向全部已连接客户端实时发送最新的报警
        BroadCastInfo(r.value());
        ack.acknowledge();
    }

    @OnMessage//收到客户端消息后自动调用的方法,message为客户端发送过来的消息
    public void onMessage(String message, Session session) {
        System.out.println("来自客户端的消息"+message);
        if (StringUtils.isNumeric(message)) {
            ArrayList<String> alarmListById = new ArrayList<>();
            Integer sensorId = Integer.parseInt(message);
            for (String alarm : alarmList) {
                if (sensorId.equals(JSON.parseObject(alarm, Alarm.class).getSensor().getId())) {
                    alarmListById.add(alarm);
                }
            }
            SendMessage(session, alarmListById.toString());
        }
    }

    //发送消息，注意：每次浏览器刷新，session会更换。
    public static void SendMessage(Session session, String message) {
        try {
            //发送给浏览器
            session.getBasicRemote().sendText("From Server:"+message+",Session ID="+session.getId());
        } catch (IOException e) {
            System.out.println("发送消息出错："+e.getMessage());
        }
    }

    //群发消息
    public static void BroadCastInfo(String message) throws IOException {
        for (Session session : SessionSet) {
            System.out.println(session.isOpen());
            if(session.isOpen()){
                SendMessage(session, message);
            }
        }
    }

    @OnClose//连接关闭调用的方法
    public void onClose(Session session) {
        SessionSet.remove(session);
    }


}
